package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {


    /**
      * 1、创建spark上下文对象
      * 2、读取数据
      *
      */

    //spark配置文件对象
    val conf = new SparkConf()

    //指定spark程序名
    conf.setAppName("Demo1WordCount")

    //设置spark程序运行模式    local:本地测试模式
    conf.setMaster("local")

    //创建spark上下文对象   spark api 的入口
    val sc = new SparkContext(conf)


    //读取本地文件    textFile可以读取本地文件也可以读取hdfs文件
    val liensRDD: RDD[String] = sc.textFile("spark/data/words.txt")

    //将数据扁平化
    val wordRDD: RDD[String] = liensRDD.flatMap(line => line.split(","))

    //将rdd转换成KV格式
    val kvRDD: RDD[(String, Int)] = wordRDD.map(word => (word, 1))


    /**
      * reduceByKey  对key进行聚合操作   只有kv格式的RDD才有
      * 1、先对key进行分组
      * 2、组内进行聚合操作
      *
      */
    val countRDD: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)

    //整理格式
    val reRD = countRDD.map(kv => kv._1 + "\t" + kv._2)


    //保存数据  将数据保存到本地或者hdfs
    reRD.saveAsTextFile("spark/data/count")


  }
}
